package lib.paxos;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import lib.ArgumentValidator;
import lib.FileIO;
import lib.Utility;
import lib.logging.ILogger;
import lib.messaging.NetworkAddress;

/**
 * State maintained by Acceptor in a Paxos group
 * @author abhip
 *
 */
public class AcceptorState implements IAcceptorState
{
    private AcceptMessage acceptedMessage;
    private List<RoundInfo> allKnownRoundInfos;
    private int currentRoundID;
    private SequenceID highestKnownSeqID;
    private ILogger logger;
    private List<NetworkAddress> nextEpochServerAddresses;
    private List<NetworkAddress> otherServerAddresses;
    private NetworkAddress serverAddress;

    public AcceptorState(
            NetworkAddress serverAddress,
            ILogger logger,
            List<NetworkAddress> otherServerAddresses)
    {
        this(
                serverAddress,
                0,
                null,
                null,
                new ArrayList<RoundInfo>(),
                logger,
                otherServerAddresses);
    }

    public AcceptorState(
            NetworkAddress serverAddress,
            int currentRoundID,
            SequenceID highestKnownSeqID,
            AcceptMessage acceptedMessage,
            List<RoundInfo> allKnownRoundInfos,
            ILogger logger,
            List<NetworkAddress> otherServerAddresses)
    {
        ArgumentValidator.IsNotNull("serverAddress", serverAddress);
        // currentRoundID = 0 means initial state, actual rounds should begin from 1
        ArgumentValidator.IsGreaterThanOrEqualTo("currentRoundID", currentRoundID, 0);
        // highestKnownSeqID can be null
        ArgumentValidator.IsNotNull("allKnownRoundInfos", allKnownRoundInfos);
        ArgumentValidator.IsNotNull("logger", logger);
        ArgumentValidator.IsNotNull("otherServerAddresses", otherServerAddresses);
        ArgumentValidator.IsGreaterThan("otherServerAddresses.size", otherServerAddresses.size(), 0);

        this.serverAddress = serverAddress;
        this.currentRoundID = currentRoundID;
        this.highestKnownSeqID = highestKnownSeqID;
        this.acceptedMessage = acceptedMessage;
        this.allKnownRoundInfos = allKnownRoundInfos;
        this.logger = logger;
        this.otherServerAddresses = otherServerAddresses;
        this.nextEpochServerAddresses = null;
    }

    @Override
    public AcceptMessage AcceptedMessage()
    {
        return acceptedMessage;
    }

    @Override
    public void AddServerToListOfKnownServers(NetworkAddress newServerAddress)
    {
        for (int i = 0; i < this.otherServerAddresses.size(); ++i)
        {
            if (newServerAddress.equals(this.serverAddress))
            {
                logger.LogError(
                        "Ignoring join request for adding current server to paxos group. Address:%s",
                        newServerAddress);
                return;
            }

            if (newServerAddress.equals(this.otherServerAddresses.get(i)))
            {
                logger.LogError(
                        "Ignoring join request by a server already in current paxos group. Address:%s",
                        newServerAddress);
                return;
            }
        }

        this.otherServerAddresses.add(newServerAddress);
    }

    @Override
    public int CurrentRoundID()
    {
        return currentRoundID;
    }

    @Override
    public void DumpAcceptorStateToFile(File outputFile) throws IOException
    {
        FileIO.OverwriteFile(outputFile, this.toString());
    }

    @Override
    public List<NetworkAddress> GetOtherServerAddresses()
    {
        return otherServerAddresses;
    }

    @Override
    public RoundInfo GetRoundInfo(int roundID) throws PaxosException
    {
        ArgumentValidator.IsBetweenBothInclusive("roundID", roundID, 1, currentRoundID);

        if((roundID == currentRoundID) && !IsCommandChosenForCurrentRound())
        {
            throw new PaxosException(
                    String.format(
                            "Round %d is in progress. No command has been chosen for it yet.",
                            roundID));
        }

        ValidateAcceptorState();
        return allKnownRoundInfos.get(roundID - 1);
    }

    @Override
    public int GetTotalNumberOfCommands()
    {
        return allKnownRoundInfos.size();
    }

    @Override
    public SequenceID HighestKnownSeqID()
    {
        return highestKnownSeqID;
    }

    @Override
    public boolean IsCommandChosenForCurrentRound() throws PaxosException
    {
        ValidateAcceptorState();

        if ((currentRoundID > 0) && (currentRoundID == allKnownRoundInfos.size()))
        {
            return true;
        }

        return false;
    }


    @Override
    public boolean IsCommandDecidedForRound(int roundID) throws PaxosException
    {
        ArgumentValidator.IsGreaterThan("roundID", roundID, 0);

        ValidateAcceptorState();

        if ((roundID <= allKnownRoundInfos.size()))
        {
            return true;
        }

        return false;
    }

    @Override
    public List<NetworkAddress> NextEpochServerAddresses()
    {
        return nextEpochServerAddresses;
    }

    @Override
    public void SetNextEpochServerAddresses(List<NetworkAddress> nextEpochServerAddresses)
    {
        // nextEpochServerAddresses can be null
        this.nextEpochServerAddresses = nextEpochServerAddresses;
    }

    @Override
    public void SetOtherServerAddresses(List<NetworkAddress> otherServerAddresses)
    {
        ArgumentValidator.IsNotNull("otherServerAddresses", otherServerAddresses);
        ArgumentValidator.IsGreaterThan("otherServerAddresses.size", otherServerAddresses.size(), 0);

        this.otherServerAddresses = otherServerAddresses;
    }

    @Override
    public String toString()
    {
        List<NetworkAddress> allServers = new ArrayList<NetworkAddress>(otherServerAddresses);
        StringBuilder sb = new StringBuilder(
                String.format(
                        "current server: %s, \n" +
                            "all servers:%s, \n" +
                            "next epoch servers:%s, \n" +
                            "currentRoundID:%d, \n" +
                            "highestKnownSeqID:%s, \n" +
                            "acceptedMessage:%s, \n" +
                            "allKnownRoundInfosCount:%d, \n" +
                            "allKnownRoundInfos:%s",
                        serverAddress,
                        Utility.InsertElementAndSort(serverAddress, allServers),
                        Utility.Sort(nextEpochServerAddresses),
                        currentRoundID,
                        highestKnownSeqID,
                        acceptedMessage,
                        allKnownRoundInfos.size(),
                        Utility.FlattenListToString(allKnownRoundInfos)));
        return sb.toString();
    }

    @Override
    public void UpdateAcceptorState(
            int currentRoundID,
            SequenceID highestKnownSeqID,
            AcceptMessage acceptedMessage,
            List<RoundInfo> missingRoundInfos) throws PaxosException
    {
        ArgumentValidator.IsGreaterThan("currentRoundID", currentRoundID, 0);

        // new round is begun only on seeing a prepare message,
        // hence highestKnownSeqID cannot be null
        ArgumentValidator.IsNotNull("highestKnownSeqID", highestKnownSeqID);

        // acceptedMessage can be null

        // missingRoundInfos can have size 0, but it cannot be null
        ArgumentValidator.IsNotNull("missingRoundInfos", missingRoundInfos);

        if (currentRoundID < this.currentRoundID)
        {
            logger.LogError(
                    "Ignoring update request, " +
                        "since proposer's roundID:%d is less than acceptor's roundID:%d, acceptor address:%s",
                    currentRoundID,
                    this.currentRoundID,
                    this.serverAddress);
            return;
        }

        this.currentRoundID = currentRoundID;
        this.highestKnownSeqID = highestKnownSeqID;
        this.acceptedMessage = acceptedMessage;
        this.allKnownRoundInfos.addAll(missingRoundInfos);
    }

    private void ValidateAcceptorState() throws PaxosException
    {
        if (currentRoundID == 0)
        {
            // initial state
            return;
        }

        int allKnownRoundInfosCount = allKnownRoundInfos.size();

        if (!(allKnownRoundInfosCount == currentRoundID ||
                allKnownRoundInfosCount == (currentRoundID - 1)))
        {
            throw new PaxosException(
                    String.format(
                            "Server has invalid state! currentRoundID:%d, allKnownRoundInfosCount:%d, they should differ by atmost 1",
                            currentRoundID,
                            allKnownRoundInfosCount));
        }
    }
}
